package com.wzw.springboot.principle.use.batch;

import com.wzw.springboot.principle.use.batch.simple.Person;
import com.wzw.springboot.principle.use.batch.simple.PersonJobExecutionListener;
import com.wzw.springboot.principle.use.batch.simple.PersonListenerSuppor;
import com.wzw.springboot.principle.use.batch.simple.PersonProcessor;
import lombok.RequiredArgsConstructor;
import org.springframework.batch.core.Job;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing;
import org.springframework.batch.core.configuration.annotation.JobBuilderFactory;
import org.springframework.batch.core.configuration.annotation.StepBuilderFactory;
import org.springframework.batch.core.launch.support.RunIdIncrementer;
import org.springframework.batch.item.database.BeanPropertyItemSqlParameterSourceProvider;
import org.springframework.batch.item.database.JdbcBatchItemWriter;
import org.springframework.batch.item.database.builder.JdbcBatchItemWriterBuilder;
import org.springframework.batch.item.file.FlatFileItemReader;
import org.springframework.batch.item.file.builder.FlatFileItemReaderBuilder;
import org.springframework.batch.item.file.mapping.BeanWrapperFieldSetMapper;
import org.springframework.context.ApplicationContext;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.ComponentScan;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Lazy;
import org.springframework.core.io.ClassPathResource;
import org.springframework.jdbc.datasource.SingleConnectionDataSource;

import javax.sql.DataSource;

/**
 * spring batch configuration
 *
 * @author Wangzhiwen
 */
@Configuration
@EnableBatchProcessing(modular = true)
@RequiredArgsConstructor
@ComponentScan(basePackages = {"com.wzw.springboot.principle.use.batch.reader", "com.wzw.springboot.principle.use.batch.writer"})
public class BatchConfiguration {

    private final JobBuilderFactory jobBuilderFactory;

    private final StepBuilderFactory stepBuilderFactory;

    private final ApplicationContext applicationContext;

    @Bean
    @Lazy
    public DataSource batchJobDataSource() {
        SingleConnectionDataSource batchJobDataSource = new SingleConnectionDataSource();
        batchJobDataSource.setUrl("jdbc:mysql://192.168.247.128:3306/batch");
        batchJobDataSource.setDriverClassName("com.mysql.cj.jdbc.Driver");
        batchJobDataSource.setUsername("root");
        batchJobDataSource.setPassword("123456");
        batchJobDataSource.setAutoCommit(true);
        return batchJobDataSource;
    }

    @Bean
    public Job simplePersonProcessJob() {
        return this.jobBuilderFactory.get("simplePersonProcessJob")
                // JobParametersIncrementer ID自增长
                .incrementer(new RunIdIncrementer())
                // 读、处理、写的监听器
                .listener(new PersonListenerSuppor())
                // JobExecutionListener
                .listener(new PersonJobExecutionListener())
                .flow(simplePersonStep())
                .end()
                .build();
    }

    @Bean
    public Step simplePersonStep() {
        // reader
        BeanWrapperFieldSetMapper<Person> mapper = new BeanWrapperFieldSetMapper<>();
        mapper.setTargetType(Person.class);
        FlatFileItemReader<Person> itemReader = new FlatFileItemReaderBuilder<Person>().name("personCsvReader")
                .resource(new ClassPathResource("person.csv"))
                .delimited().names("name", "age")
                .fieldSetMapper(mapper)
                .build();

        // writer
        BeanPropertyItemSqlParameterSourceProvider<Person> parameterSourceProvider = new BeanPropertyItemSqlParameterSourceProvider<>();
        // JdbcBatchItemWriter 实现了InitializingBean接口。要放到容器中才会初始化内部的值
        JdbcBatchItemWriter<Person> jdbcBatchItemWriter = new JdbcBatchItemWriterBuilder<Person>()
                // .itemSqlParameterSourceProvider(parameterSourceProvider)
                // 自动适配Bean对象映射到SQL语句
                .beanMapped()
                // .itemPreparedStatementSetter((item, ps) -> {
                //     ps.setString(1, item.getName());
                //     ps.setInt(2, item.getAge());
                // })
                // 自动适配PreparedStatement映射到占位符中
                // .columnMapped()
                .sql(ColumnHandler.mappedWithInsert(Person.class))
                .dataSource(batchJobDataSource())
                .build();
        // 不放入容器手动初始化
        jdbcBatchItemWriter.afterPropertiesSet();

        return stepBuilderFactory.get("simplePersonStep")
                .<Person, Person>chunk(10)
                .reader(itemReader)
                .processor(new PersonProcessor())
                .writer(jdbcBatchItemWriter)
                .build();
    }
}
